package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Peize
 * @date 2024/10/20
 *      dwd加购事实表   flink SQL
 */
public class DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(
                10013,
                4,
                Constant.TOPIC_DWD_TRADE_CART_ADD
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // TODO: 2024/10/20 1.工具类跟kafka获取sql连接器  然后直接从kafka主题区里读走数据      流->动态表
        readOdsDB(tableEnv,Constant.TOPIC_DWD_TRADE_CART_ADD);
        // tableEnv.executeSql("select * from topic_db").print();
        // TODO: 2024/10/20 2.从读来的大动态表中过滤 把加购表找出来
        Table cartInfo = tableEnv.sqlQuery(
                "select \n" +
                        "    `data`['id'] id,\n" +
                        "    `data`['user_id'] user_id,\n" +
                        "    `data`['course_id'] course_id,\n" +
                        "    `data`['cart_price'] cart_price,\n" +
                       // "    `data`['chapter_id'] chapter_id,\n" +
                       // "    `data`['comment_txt'] comment_txt,\n" +
                        "    `data`['sold'] sold,\n" +
                        "    ts\n" +
                        "from topic_db \n" +
                        "where `table`='cart_info'\n" +
                        "    and (\n" +
                        "        `type` = 'insert' \n" +
                        "        or\n" +
                        "        `type` = 'update')"
                        // "and `type` = 'insert'"
        );
        // cartInfo.execute().print();

        // TODO: 2024/10/20 3.把加购行为写入kafka主题中
        // 3.1 建一个动态表   映射给kafka主题
        tableEnv.executeSql(
                        "create table "+Constant.TOPIC_DWD_TRADE_CART_ADD+"(\n" +
                        "    id string,\n" +
                        "    user_id string,\n" +
                        "    course_id string,\n" +
                        "    cart_price string,\n" +
                        // "    chapter_id string,\n" +
                        // "    comment_txt string,\n" +
                        "    sold string,\n" +
                        "    ts bigint,\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED\n" +
                        ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));
        //  3.2 往映射表里写入数据!!!
        cartInfo.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);
    }
}
